package temp.casia.isiteam.zrb.hbase.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import temp.casia.isiteam.zrb.hbase.client.bean.HbaseStatus;
import temp.casia.isiteam.zrb.hbase.client.util.Commons;
import temp.casia.isiteam.zrb.hbase.client.util.ConnectionManager;

import com.hanhai.zrb.model.project.ProjectDetail;

public class ProjectDetailHbaseClient {
	
	public final static String TABLE_NAME = "project_basic";
	
	protected Configuration config;
	private Logger logger = Logger.getLogger(ProjectDetailHbaseClient.class);
	/**
	 * 表名
	 */
	protected Map<String, TableName> tableNameMap;
	protected String quorum;
	protected Integer currentClientNum = 0;
	private Timer timer;
	private int retryTimes = 0;
	
	/**
	 * 使用默认的master地址
	 */
	public ProjectDetailHbaseClient(){
		this.quorum = Commons.quorum;
		this.initConfig();
	}
	


	
	/**
	 * 指定master地址
	 * @param quorum
	 */
	public ProjectDetailHbaseClient(String quorum){
		this.quorum = quorum;
		this.initConfig();
	}
	
	protected void initConfig() {
		Connection conn = null;
		try {
			config = HBaseConfiguration.create();
			config.set("hbase.zookeeper.quorum", this.quorum);
			config.set("hbase.client.retries.number", "3");
			config.set("zookeeper.recovery.retry", "3");
			
			conn = ConnectionManager.getConn(config);
			Admin admin = conn.getAdmin();
			TableName[] tableAry = admin.listTableNames("project_.*");
			this.tableNameMap = new HashMap<String, TableName>();
			for (TableName tableName : tableAry) {
				this.tableNameMap.put(tableName.getNameAsString(), tableName);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(conn);
		}
	}

	protected Connection initConn() {
		Connection conn = ConnectionManager.getConn(config);
		return conn;
	}
	
	private void closeConn(Connection conn){
		try {
			conn.close();
		} catch (Exception e) {e.printStackTrace();}
	}
	
	private void closeConn(Table table,Connection conn){
		try {
			table.close();
			conn.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	
	/**
	 * 赋值
	 * @param put
	 * @param famByt
	 * @param qualifier
	 * @param value
	 */
	private void addValue(Put put, byte[] famByt, String qualifier, String value){
		if(value != null){
			put.addColumn(famByt, Bytes.toBytes(qualifier), Bytes.toBytes(value));
		}
	}
	
	/**
	 * 获取指定列的值
	 * @param result
	 * @param familly
	 * @param rowName
	 * @return
	 */
	private String getRow(Result result,String familly,String rowName){
		byte[] byts = result.getValue(Bytes.toBytes(familly), Bytes.toBytes(rowName));
		String row = null;
		if (byts != null && byts.length > 0) {
			row = Bytes.toString(byts);
		}
		return row;
	}
	
	public HbaseStatus insertProjectDetail(ProjectDetail projectDetail){
		HbaseStatus status = HbaseStatus.successed;
		Table table = null;
		Connection conn = null;
		try {
			conn = this.initConn();
			table = conn.getTable(this.tableNameMap.get(TABLE_NAME));
			String rowKey = String.valueOf(projectDetail.getId());
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes("basic");
			
			this.addValue(put, famByt, "intro", projectDetail.getIntro());
			this.addValue(put, famByt, "address", projectDetail.getAddress());
			this.addValue(put, famByt, "geocoor", projectDetail.getGeocoor());
			this.addValue(put, famByt, "destionation", projectDetail.getDestionation());
			this.addValue(put, famByt, "advantage", projectDetail.getAdvantage());
			this.addValue(put, famByt, "plan", projectDetail.getPlan());
			this.addValue(put, famByt, "company", projectDetail.getCompany());
			this.addValue(put, famByt, "product", projectDetail.getProduct());
			this.addValue(put, famByt, "team", projectDetail.getTeam());
			this.addValue(put, famByt, "completion", projectDetail.getCompletion());
			this.addValue(put, famByt, "proMode", projectDetail.getProMode());
			
			table.put(put);
			
			this.retryTimes = 0;
		} catch (Exception e) {
			this.logger.error("Exception!", e);
			if(++this.retryTimes <= 5){
				try {
					Thread.sleep(2000);
				} catch (Exception e1) {}
				status = this.insertProjectDetail(projectDetail);
			} else{
				status = HbaseStatus.exception;
			}
		} finally{
			this.closeConn(table, conn);
		}
		return status;
	}
	
	public ProjectDetail getProjectDetail(long id){
		ProjectDetail projectDetail = null;
		Table table = null;
		Connection conn = null;
		try {
			conn = this.initConn();
			table = conn.getTable(this.tableNameMap.get(TABLE_NAME));
			String rowKey = String.valueOf(id);
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if (result != null) {
				projectDetail = this.getProjectBean(result);
			}
			this.retryTimes = 0;
		} catch (Exception e) {
			this.logger.error("Exception! key: " + id, e);
			if(++this.retryTimes <= 5){
				try {
					Thread.sleep(2000);
				} catch (Exception e1) {}
				projectDetail = this.getProjectDetail(id);
			}
		} finally {
			this.closeConn(table, conn);
		}
		return projectDetail;
	}
	
	
	/**
	 * 
	 * 批量查询项目
	 * @param termList
	 * @return
	 */
	public List<ProjectDetail> getProjectInfoBasic(List<Long> idList){
		List<ProjectDetail> list = new ArrayList<ProjectDetail>();
		Table table = null;
		Connection conn = null;
		try {
			conn = this.initConn();
			table = conn.getTable(this.tableNameMap.get(TABLE_NAME));
			List<Get> getList = new ArrayList<Get>();
			for(Long id: idList){
				if(id == null){
					this.logger.warn("id of project is null!");
					continue;
				}
				String rowKey = id.toString();
				Get get = new Get(Bytes.toBytes(rowKey));
				getList.add(get);
			}
			Result[] results = table.get(getList);
			for(Result result: results){
				ProjectDetail detail = this.getProjectBean(result);
				if(detail != null){
					list.add(detail);
				}
			}
		} catch (Exception e) {
			this.logger.error("Exception! ", e);
			if(++this.retryTimes <= 5){
				try {
					Thread.sleep(2000);
				} catch (Exception e1) {}
				list = this.getProjectInfoBasic(idList);
			}
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		return list;
	}
	
	private ProjectDetail getProjectBean(Result result){
		if(result.isEmpty()){
			return null;
		}
		ProjectDetail projectDetail = new ProjectDetail();
		String rowKey = Bytes.toString(result.getRow());
		projectDetail.setId(Long.parseLong(rowKey));
		
		projectDetail.setIntro(this.getRow(result, "basic", "intro"));
		projectDetail.setAddress(this.getRow(result, "basic", "address"));
		projectDetail.setGeocoor(this.getRow(result, "basic", "geocoor"));
		projectDetail.setDestionation(this.getRow(result, "basic", "destionation"));
		projectDetail.setAdvantage(this.getRow(result, "basic", "advantage"));
		projectDetail.setPlan(this.getRow(result, "basic", "plan"));
		projectDetail.setCompany(this.getRow(result, "basic", "company"));
		projectDetail.setProduct(this.getRow(result, "basic", "product"));
		projectDetail.setTeam(this.getRow(result, "basic", "team"));
		projectDetail.setCompletion(this.getRow(result, "basic", "completion"));
		projectDetail.setProMode(this.getRow(result, "basic", "proMode"));
		
		return projectDetail;
	}

}
